package com.xxx.service;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 原生实现rocketmq顺序消息消费者②
 */
@Service
public class OrderedMessageConsumer {

    @Value("${rocketmq.name-server}")
    private String nameServer;
    @Value("${rocketmq.producer.group}")
    private String group;

    private DefaultMQPushConsumer consumer;

    // 系统启动时初始化consumer
    @PostConstruct
    public void init() throws MQClientException {
        // 1.创建顺序消息消费者
        consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe("ordered-topic", "*");

        // 2.设置顺序消费模式
        consumer.registerMessageListener(new MessageListenerOrderly() {  // 顺序消息使用MessageListenerOrderly监听，表示只会用1个线程去对应消费1个队列
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("线程名称：【" + Thread.currentThread().getName() + "】收到顺序消息: %s%n", new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        // 3.启动顺序消息消费者
        consumer.start();
        System.out.println("顺序消费者已启动...");
    }

    // 关闭consumer
    public void shutdown() {
        if (consumer != null) {
            consumer.shutdown();
        }
    }
}
